package com.mapreduce;

import com.bean.AvgTemp;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.net.URI;

public class AvgTempMapReduce extends Configured implements Tool {
//  1992,23
//  1993,22
//  1992,34
//  1993,26
//  1994,25
//    -> keyvalueTextInputFormat
//    k1        v1
//    1992      23
//    1993      22
//    1992      34
//    1993      26
//    1994      25
    static class AvgTempMapper extends
        Mapper<Text, Text, Text, AvgTemp> {
    @Override
    protected void map(Text key, Text value, Mapper<Text, Text, Text, AvgTemp>.Context context) throws IOException, InterruptedException {
        AvgTemp avgTemp = new AvgTemp(
                1,Integer.parseInt(value.toString()));
        context.write(key,avgTemp);
    }
}
//    ->map
//    k2        v2
//    1992      AvgTemp(1,23)
//    1993      AvgTemp(1,22)
//    1992      AvgTemp(1,34)
//    1993      AvgTemp(1,26)
//    1994      AvgTemp(1,25)
//    k2        v2
//    1993      [AvgTemp(1,23),AvgTemp(1,26)]
//    1992      [AvgTemp(1,23),AvgTemp(1,34)]
//    1994      [AvgTemp(1,25)]
    static class AvgTempCombiner extends
        Reducer<Text, AvgTemp, Text, AvgTemp> {
    @Override
    protected void reduce(Text key, Iterable<AvgTemp> values, Reducer<Text, AvgTemp, Text, AvgTemp>.Context context) throws IOException, InterruptedException {
        // 提前combiner， combiner相当于reduce，
        // 但是combiner是在map所在的机器上运行的
        int num = 0;
        int sum = 0;
        for (AvgTemp val:values){
//           统计出现次数
            num += val.getNum();
//           统计温度值总和
            sum += val.getSum();
        }
        context.write(key,new AvgTemp(num,sum));
    }
}
//    ->combiner(reduce)
//    k2        v2
//    1993      [AvgTemp(1,23),AvgTemp(1,26)]
//    1992      [AvgTemp(1,23),AvgTemp(1,34)]
//    1994      [AvgTemp(1,25)]

//    k2        v2
//    1993      AvgTemp(2,49)
//    1992      AvgTemp(2,57)
//    1994      AvgTemp(1,25)

//    ->shuffle
//    ->reduce
    static class AvgTempReducer extends
        Reducer<Text, AvgTemp, Text, DoubleWritable>{
    @Override
    protected void reduce(Text key, Iterable<AvgTemp> values, Reducer<Text, AvgTemp, Text, DoubleWritable>.Context context) throws IOException, InterruptedException {
        double avg = 0.0;
        for (AvgTemp val:values){
            avg = (double) val.getSum() / val.getNum();
        }
        context.write(key,new DoubleWritable(avg));
    }
}
    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator",",");

        // 指定输入输出路径
        Path input = new Path(
                "hdfs://192.168.10.11:9000/sortdata");
        Path output = new Path(
                "hdfs://192.168.10.11:9000/avgoutput");

        FileSystem fs = FileSystem.get(
                new URI("hdfs://192.168.10.11:9000")
                ,conf);
        if (fs.exists(output)) fs.delete(output,true);

        //构建Job
        Job job = Job.getInstance(conf);
        job.setJobName("avg");
        job.setJarByClass(this.getClass());
        // 指定map类的配置
        job.setMapperClass(AvgTempMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(AvgTemp.class);
        // 设置提前combiner
        job.setCombinerClass(AvgTempCombiner.class);

        // 指定reduce类的配置
        job.setReducerClass(AvgTempReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        job.setInputFormatClass(KeyValueTextInputFormat.class);
        KeyValueTextInputFormat.addInputPath(job,input);

        TextOutputFormat.setOutputPath(job,output);

        return job.waitForCompletion(true)?0:-1;
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new AvgTempMapReduce(),args));
    }
}
